#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__

#include "buffer.hpp"
#include <thread>
#include <mutex>
#include <atomic>
#include <functional>
#include <memory>
#include <condition_variable>

namespace Alpaca
{
    enum class AsyncType
    {
        ASYNC_SAFE,  // 缓冲区满了则阻塞
        ASYNC_UNSAFE // 无限扩容
    };
    class AsyncLooper
    {
    public:
        using func_t = std::function<void(Buffer &)>;
        using ptr = std::shared_ptr<AsyncLooper>;

        AsyncLooper(func_t func, AsyncType looper_type = AsyncType::ASYNC_SAFE)
            : _stop(false),
              _looper_type(looper_type),
              _thread(std::thread(&AsyncLooper::threadEntry, this)),
              _func(func) {}

        ~AsyncLooper()
        {
            stop();
        }

        void stop()
        {
            _stop = true;
            _cond_con.notify_all();
            _thread.join(); // 等待工作线程退出
        }

        void push(const char *data, size_t len)
        {
            // 1.无线扩容  --非安全 2.固定大小

            std::unique_lock<std::mutex> lock(_mutex);
            if (_looper_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&]()
                               { return _pro_buf.WriteAbleSize() >= len; });

            // 走到这说明缓冲区中有足够空间够我们写入
            _pro_buf.push(data, len);
            _cond_con.notify_one(); // 唤醒消费者对缓冲区进行处理
        }

    private:
        // 对消费缓冲区中的数据进行处理后，初始化、再交换缓冲区
        void threadEntry() // 线程入口函数
        {
            while (1)
            {
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 避免生产缓冲区中有数据但没有被完全处理的情况
                    if (_stop && _pro_buf.empty())
                        break;
                    if (_looper_type == AsyncType::ASYNC_SAFE)
                        _cond_con.wait(lock, [&]()
                                       { return _stop || !_pro_buf.empty(); });
                    _con_buf.swap(_pro_buf);
                    _cond_pro.notify_all();
                }
                _func(_con_buf);
                _con_buf.reset();
            }
        }

    private:
        AsyncType _looper_type;
        std::atomic<bool> _stop;
        Buffer _pro_buf; // 生产者缓冲区
        Buffer _con_buf; // 消费者缓冲区
        std::mutex _mutex;
        std::condition_variable _cond_pro;
        std::condition_variable _cond_con;
        std::thread _thread; // 异步工作器对应的工作线程
        func_t _func;        // 回调函数
    };
}

#endif